package com.bblei.hbaseDemo.mapReduceDemo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.LinkedList;


public class HBaseMr {
    /**
     * 创建hbase配置
     */
    public static Configuration config = null;

    static {
        config = HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum", "slave1,slave2,slave3");
        config.set("hbase.zookeeper.property.clientPort", "2181");
    }

    /**
     * 表信息
     */
    public static final String tableName = "word";//表名1
    public static final String colf = "content";//列族
    public static final String col = "info";//列
    public static final String tableName2 = "stat";//表名2

    /**
     * 初始化表结构，及其数据
     */
    public static void initTB() {
        HTable table = null;
        HBaseAdmin admin = null;
        Connection connection = null;
        try {
//            admin = new HBaseAdmin(config);//创建表管理
           connection = ConnectionFactory.createConnection(config);
            admin = (HBaseAdmin) connection.getAdmin();
            /*删除表*/
            if (admin.tableExists(tableName) || admin.tableExists(tableName2)) {
                System.out.println("table is already exists!");
                admin.disableTable(tableName);
                admin.deleteTable(tableName);
                admin.disableTable(tableName2);
                admin.deleteTable(tableName2);
            }
            /*创建表*/
//            HTableDescriptor desc = new HTableDescriptor(tableName);
            HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
            HColumnDescriptor family = new HColumnDescriptor(colf);
            desc.addFamily(family);
            admin.createTable(desc);
//            HTableDescriptor desc2 = new HTableDescriptor(tableName2);
            HTableDescriptor desc2 = new HTableDescriptor(TableName.valueOf(tableName2));
            HColumnDescriptor family2 = new HColumnDescriptor(colf);
            desc2.addFamily(family2);
            admin.createTable(desc2);
            /*插入数据*/

            connection = ConnectionFactory.createConnection(config);
            table = (HTable) connection.getTable(TableName.valueOf(tableName));

//            table = new HTable(config, tableName);
            table.setAutoFlushTo(false);
//            table.setAutoFlush(false);
            table.setWriteBufferSize(5);
            LinkedList<Put> lp = new LinkedList<Put>();

            Put p1 = new Put(Bytes.toBytes("1"));
            p1.addColumn(Bytes.toBytes(colf), Bytes.toBytes(col), Bytes.toBytes("The Apache Hadoop software library is a framework"));
            lp.add(p1);


            Put p2 = new Put(Bytes.toBytes("2"));
            p2.addColumn(Bytes.toBytes(colf), Bytes.toBytes(col), Bytes.toBytes("The common utilities that support the other Hadoop modules"));
            lp.add(p2);


            Put p3 = new Put(Bytes.toBytes("3"));
            p3.addColumn(Bytes.toBytes(colf), Bytes.toBytes(col), Bytes.toBytes("Hadoop by reading the documentation"));
            lp.add(p3);


            Put p4 = new Put(Bytes.toBytes("4"));
            p4.addColumn(Bytes.toBytes(colf), Bytes.toBytes(col), Bytes.toBytes("Hadoop from the release page"));
            lp.add(p4);


            Put p5 = new Put(Bytes.toBytes("5"));
            p5.addColumn(Bytes.toBytes(colf), Bytes.toBytes(col), Bytes.toBytes("Hadoop on the mailing list"));
            lp.add(p5);

            table.put(lp);
            table.flushCommits();
            lp.clear();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (table != null) {
                    table.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }
}
